/////////////////////////////////////////////////////////////////////////////
// Original code from libhdfs3. Copyright (c) 2013 - 2014, Pivotal Inc.
// All rights reserved. Author: Zhanwei Wang
/////////////////////////////////////////////////////////////////////////////
//  Modifications by Kumo Inc.
// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//



#include <string>
#include <limits>

#include <kmhdfs/common/exception.h>
#include <kmhdfs/common/exception_internal.h>
#include <kmhdfs/common/file_wrapper.h>

namespace Hdfs {
    namespace Internal {
        CFileWrapper::CFileWrapper() : file(nullptr) {
        }

        CFileWrapper::~CFileWrapper() {
            close();
        }

        bool CFileWrapper::open(int fd, bool delegate) {
            int newfd = fd;

            if (!delegate) {
                newfd = dup(fd);

                if (newfd < 0) {
                    THROW(HdfsIOException, "Cannot duplicate file descriptor: %s",
                          GetSystemErrorInfo(errno));
                }
            }

            file = fdopen(newfd, "rb");

            if (nullptr == file && !delegate) {
                ::close(newfd);
            }

            return nullptr != file;
        }

        bool CFileWrapper::open(const std::string &path) {
            this->path = path;
            file = fopen(path.c_str(), "rb");
            return nullptr != file;
        }

        void CFileWrapper::close() {
            if (nullptr != file) {
                fclose(file);
                file = nullptr;
            }
        }

        const char *CFileWrapper::read(std::vector<char> &buffer, int32_t size) {
            buffer.resize(size);
            copy(buffer.data(), size);
            return buffer.data();
        }

        void CFileWrapper::copy(char *buffer, int32_t size) {
            int32_t todo = size, done;

            while (todo > 0) {
                done = fread(buffer + (size - todo), sizeof(char), todo, file);

                if (done < 0) {
                    THROW(HdfsIOException, "Cannot read file \"%s\", %s.", path.c_str(),
                          GetSystemErrorInfo(errno));
                } else if (0 == done) {
                    THROW(HdfsIOException, "Cannot read file \"%s\", End of file.",
                          path.c_str());
                }

                todo -= done;
            }
        }

        void CFileWrapper::seek(int64_t offset) {
            assert(offset >= 0);
            int64_t todo = offset, batch;
            bool seek_set = true;

            do {
                batch = todo < std::numeric_limits<long>::max()
                            ? todo
                            : std::numeric_limits<long>::max();
                off_t rc = fseek(file, static_cast<long>(batch),
                                 seek_set ? SEEK_SET : SEEK_CUR);
                seek_set = false;

                if (rc != 0) {
                    THROW(HdfsIOException, "Cannot lseek file: %s, %s", path.c_str(),
                          GetSystemErrorInfo(errno));
                }

                todo -= batch;
            } while (todo > 0);
        }
    }
}
